Add pubsub support to the Glide driver by liorsve · Pull Request #44 · valkey-io/spring-data-valkey

liorsve · 2026-01-12T15:03:20Z

This PR adds pubsub support to the spring-data-valkey glide driver (implements ValkeyGlideSubscriptions and the pubsub API of ValkeyGlideConnectio).
It does so by setting the glide clients with a pubsub callback, which is later routing messages to the pubsub listeners provided by the app.

jduo · 2026-01-15T17:48:11Z

...java/io/valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideConnection.java

-                // () -> nativeClient.customCommand(new String[]{"SUNSUBSCRIBE"}).get()
+                // This is defensive -  subscriptions were already supposed to be 
+                // cleaned during subscription.close()
+                () -> nativeClient.customCommand(new String[]{"UNSUBSCRIBE_BLOCKING"}).get(),


@liorsve @ikolomi
This can have a big impact on performance as we're executing 3 extra requests when returning a connection to the pool.

We saw when implement #46 that when running the performance test, the UNWATCH request alone halved the speed of the connector.

Ideally we make these unsubscribe calls conditional. At the very least we should batch them (and probably the UNWATCH if it is needed) together or block on the returned futures all-at-once.

IMO we can remove this entirely if this makes such an impact, as this happens also at subscription.close(). We call it there with theUNSUBSCRIBE_BLOCKINGoption which means that we will wait for server confirmation of unsubscribing before returning the connection to the pull. Removing this

jeremyprime · 2026-01-15T23:35:09Z

.../springframework/data/valkey/connection/valkeyglide/ValkeyGlideDelegatingPubSubListener.java

+    boolean hasListener() {
+        return messageListener != null;
+    }
+}


Nit: No ending newlines in new classes.

jeremyprime · 2026-01-16T00:01:39Z

...va/io/valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideSubscription.java

+        // Unsubscribe from SPECIFIC channels we subscribed to, not ALL
+        if (channelsToNotify.length > 0) {
+            try {
+                sendPubsubCommand("UNSUBSCRIBE_BLOCKING");


Similar to connection cleanup, if possible would be good to batch the unsubscribe and punsubscribe calls to avoid multiple network calls.

I'm afraid with the new dynamic pubsub impl in glide it's not really possible - We added a pubsub synchronizer in the core that manages subscriptions and subscription confirmation from the server, which doesn't support sending them as a batch. If we would try to bypass that logic with sending custom commands in a batch, the client would get the unsubscribe push notifications from the server and would think it got unsubscribed wrongfully, and so would try to resubscribe. If this is so impactful in terms of performance we can think of adding this functionality to the glide synchronizer in the future. But now that I removed unsubscribe calls on each connection cleanup, the subscription cleanup only happens on a connection that had an active subscription when closing. And if people are using the Listener Container as the spring docs advise, this container only uses 1 connection for all subscriptions in its entire lifetime, so this cleanup will be called once only when the container is destroyed.

Lets not hide the exceptions

jeremyprime · 2026-01-16T16:44:04Z

pom.xml

 		<xstream>1.4.21</xstream>
 		<pool>2.11.1</pool>
-		<valkey.glide>2.1.1</valkey.glide>
+		<valkey.glide>2.3.0-rc2</valkey.glide>


For Beta this is fine, I assume non-RC will be available by the time we go GA.

This also means we will get the Windows changes (move to JNI) from 2.2.0, which should be tested more thoroughly at some point. Although nothing more to do in this PR.

Its ok for betta

jeremyprime

Other than concern over performance in connection and subscription cleanup, looks good to me.

Signed-off-by: Lior Sventitzky <liorsve@amazon.com> try 2 - callback in config Signed-off-by: Lior Sventitzky <liorsve@amazon.com> added listener map Signed-off-by: Lior Sventitzky <liorsve@amazon.com> changed connection to use AbstractSubscription methods Signed-off-by: Lior Sventitzky <liorsve@amazon.com> enabled glide on interfacr tests, added glude specific tests Signed-off-by: Lior Sventitzky <liorsve@amazon.com> removed more debug Signed-off-by: Lior Sventitzky <liorsve@amazon.com> fixed to unifiedClient Signed-off-by: Lior Sventitzky <liorsve@amazon.com> fix Signed-off-by: Lior Sventitzky <liorsve@amazon.com> reverted to non arg unsusbcribes, removed stub subscription in config (won't work with old glide) Signed-off-by: Lior Sventitzky <liorsve@amazon.com> started cleanup for PR Signed-off-by: Lior Sventitzky <liorsve@amazon.com> added blocking to subscription commands Signed-off-by: Lior Sventitzky <liorsve@amazon.com> change to rc version, remove protobuf dependency Signed-off-by: Lior Sventitzky <liorsve@amazon.com>

Signed-off-by: Lior Sventitzky <liorsve@amazon.com>

ikolomi

Approved pending comments resolution

ikolomi · 2026-01-18T13:04:21Z

.../valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideClusterConnection.java


-    public ValkeyGlideClusterConnection(ClusterGlideClientAdapter clusterAdapter, @Nullable ValkeyGlideConnectionFactory factory) {
-        this(clusterAdapter, factory, Duration.ofMillis(100));
+    public ValkeyGlideClusterConnection(ClusterGlideClientAdapter clusterAdapter, 


Just appears this way on github, see + signs beside

ikolomi · 2026-01-18T13:09:35Z

.../valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideClusterConnection.java

+    public ValkeyGlideClusterConnection(ClusterGlideClientAdapter clusterAdapter, 
+            @Nullable ValkeyGlideConnectionFactory factory,
+            @Nullable DelegatingPubSubListener pubSubListener,
+            Duration cacheTimeout) {


why not as the last param?

ikolomi · 2026-01-18T13:10:20Z

.../valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideClusterConnection.java


-    public ValkeyGlideClusterConnection(ClusterGlideClientAdapter clusterAdapter, @Nullable ValkeyGlideConnectionFactory factory, Duration cacheTimeout) {
-        super(clusterAdapter, factory);
+    public ValkeyGlideClusterConnection(ClusterGlideClientAdapter clusterAdapter, 


Just looks like it bc of github

ikolomi · 2026-01-18T13:46:47Z

.../valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideConnectionFactory.java

    private boolean earlyStartup = true;
    private int phase = 0;

+    private final Map<Object, DelegatingPubSubListener> clientListenerMap = new ConcurrentHashMap<>();


please comment what is the Object and what for

ikolomi · 2026-01-18T13:53:39Z

...java/io/valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideConnection.java


    protected final UnifiedGlideClient unifiedClient;
    protected final @Nullable ValkeyGlideConnectionFactory factory;
+    protected final @Nullable DelegatingPubSubListener pubSubListener;


comment why do we "Delegate" (we are using glide's callback which is supplied on the creation , yada yda)

ikolomi · 2026-01-18T14:18:10Z

...va/io/valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideSubscription.java

+    /**
+     * Send a pub/sub command directly to the client using GlideString.
+     */
+    private void sendPubsubCommand(String command, byte[]... args) {


args -> channels

ikolomi · 2026-01-18T14:20:49Z

...va/io/valkey/springframework/data/valkey/connection/valkeyglide/ValkeyGlideSubscription.java

+
+        try {
+            client.customCommand(glideArgs);
+        } catch (InterruptedException e) {


one case is cleaner

ikolomi · 2026-01-18T14:27:27Z

...ngframework/data/valkey/connection/valkeyglide/ValkeyGlidePubSubPoolingIntegrationTests.java

+        String channelBase = "test:concurrent:isolation";
+
+        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+        CountDownLatch allDone = new CountDownLatch(threadCount);


allDone probably not required - use
executorService.shutdown();
executorService.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS);

ikolomi · 2026-01-18T14:32:02Z

...ta-valkey/src/test/java/io/valkey/springframework/data/valkey/listener/PubSubTestParams.java

 import io.valkey.springframework.data.valkey.connection.lettuce.LettuceConnectionFactory;
 import io.valkey.springframework.data.valkey.connection.lettuce.extension.LettuceConnectionFactoryExtension;
+import io.valkey.springframework.data.valkey.connection.valkeyglide.ValkeyGlideConnectionFactory;
+import io.valkey.springframework.data.valkey.connection.valkeyglide.extension.ValkeyGlideConnectionFactoryExtension;


not really there just appears this way

ikolomi · 2026-01-18T14:32:39Z

...ta-valkey/src/test/java/io/valkey/springframework/data/valkey/listener/PubSubTestParams.java

 		return parameters;
 	}

+


Signed-off-by: Lior Sventitzky <liorsve@amazon.com>

liorsve force-pushed the spring-pubsub branch 5 times, most recently from 54c2f6b to c62cee3 Compare January 15, 2026 09:00

liorsve marked this pull request as ready for review January 15, 2026 09:12

liorsve changed the title ~~Spring pubsub~~ Add pubsub to the glide driver Jan 15, 2026

jduo reviewed Jan 15, 2026

View reviewed changes

jeremyprime reviewed Jan 15, 2026

View reviewed changes

jeremyprime reviewed Jan 16, 2026

View reviewed changes

jeremyprime approved these changes Jan 16, 2026

View reviewed changes

jeremyprime linked an issue Jan 16, 2026 that may be closed by this pull request

Integrate reliable and dynamic PubSub from Valkey-Glide #11

Closed

liorsve added 2 commits January 18, 2026 12:23

removed defensive unsubscribe, lint fix

57cd164

Signed-off-by: Lior Sventitzky <liorsve@amazon.com>

liorsve force-pushed the spring-pubsub branch from b451ba2 to 57cd164 Compare January 18, 2026 12:36

ikolomi approved these changes Jan 18, 2026

View reviewed changes

ikolomi assigned liorsve Jan 18, 2026

ikolomi added the enhancement New feature or request label Jan 18, 2026

ikolomi added this to the Beta 0.2.0 milestone Jan 18, 2026

addressed comments

aea2534

Signed-off-by: Lior Sventitzky <liorsve@amazon.com>

liorsve force-pushed the spring-pubsub branch from 2c2879f to aea2534 Compare January 18, 2026 15:53

liorsve changed the title ~~Add pubsub to the glide driver~~ Add pubsub support to the Glide driver Jan 18, 2026

liorsve merged commit f4529e4 into valkey-io:main Jan 19, 2026
51 of 52 checks passed

Conversation

liorsve commented Jan 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jduo Jan 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

liorsve Jan 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jeremyprime left a comment

Choose a reason for hiding this comment

Uh oh!

ikolomi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

liorsve Jan 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

liorsve commented Jan 12, 2026 •

edited

Loading

jduo Jan 15, 2026 •

edited

Loading

liorsve Jan 18, 2026 •

edited

Loading

liorsve Jan 18, 2026 •

edited

Loading